package com.example.ssedemo.service;

import com.example.ssedemo.InboxUserTopicListener;
import com.example.ssedemo.bean.Message;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

@Slf4j
@Component
@Data
public class SseEmitterManager {

    // 存储每个客户端连接的SSE Emitter实例
    public static final Map<String, SseEmitter> EMITTERS = new ConcurrentHashMap<>();

    private static final String HEARTBEAT_TYPE = "heartbeat";

    private static final String HEARTBEAT_MESSAGE = "heartbeat";

    // 使用线程池管理心跳任务
    private final ScheduledExecutorService heartbeatExecutors = Executors.newScheduledThreadPool(1);
    @Resource
    private InboxUserTopicListener listener;
    @Resource
    private RedissonClient redissonClient;
    @Value("${server.port}")
    private Integer port;

    public SseEmitter connect(String id) {
        //安全的从EMITTERS中移除旧的SSE Emitter
        EMITTERS.compute(id, (k, v) -> {
            if (v != null) {
                v.complete();
            }
            return null;
        });
        SseEmitter sseEmitter = new SseEmitter(0L);
        sseEmitter.onCompletion(completionCallBack(id));
        sseEmitter.onTimeout(timeoutCallBack(id));
        sseEmitter.onError(errorCallBack(id));
        EMITTERS.put(id, sseEmitter);

        RTopic topic = redissonClient.getTopic(topicName(id));

        if (topic.countListeners() == 0) {
            //添加监听器
            topic.addListener(Message.class, listener);
        }

        startHeartbeat(id);

        log.info("连接建立,当前客户端数量:{}", EMITTERS.size());
        return sseEmitter;
    }

    private void startHeartbeat(String id) {
        Runnable sendHeartbeat = () -> {
            try {
                heartbeat(id);
            } catch (IOException e) {
                log.error("心跳发送失败，移除连接: {}", id, e);
                remove(id);
                // 必要时关闭SseEmitter或其他资源
            }
        };
        // 定时执行心跳发送任务，每10秒一次
        heartbeatExecutors.scheduleAtFixedRate(sendHeartbeat, 0L, 10, TimeUnit.SECONDS);
    }

    private void send(Message message) throws IOException {
        String id = message.getId();
        EMITTERS.get(id).send(message);
    }

    //完成回调
    private Runnable completionCallBack(String id) {
        return () -> remove(id);
    }

    //异常回调
    private Consumer<Throwable> errorCallBack(String id) {
        return throwable -> remove(id);
    }

    //超时回调
    private Runnable timeoutCallBack(String id) {
        return () -> remove(id);
    }

    private void remove(String id) {
        try {
            log.info("用户 {} 连接关闭,剩余客户端数量:{}", id, EMITTERS.size());
            SseEmitter remove = EMITTERS.remove(id);
            if (remove != null) {
                remove.complete();
            }
            // 移除监听器
            RTopic topic = redissonClient.getTopic(topicName(id));
            topic.removeAllListeners();
        } catch (Exception e) {
            log.error("移除操作中发生异常", e);
        }
    }

    // topic名称
    private String topicName(String id) {
        return String.format("message-%s", id);
    }

    private void heartbeat(String id) throws IOException {
        Message message = new Message(id, HEARTBEAT_TYPE, "消息来自于:" + port + HEARTBEAT_MESSAGE);
        send(message);
    }

    public void publish(Message message) {
        String id = message.getId();
        RTopic topic = redissonClient.getTopic(topicName(id));
        topic.publish(message);
    }

}
